package club.janna.netty.qps;

import io.netty.bootstrap.Bootstrap;
import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandler;
import io.netty.channel.ChannelHandlerContext;
import io.netty.channel.ChannelInboundHandlerAdapter;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;

import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;

public class Client {
    static final int N = 1000; // 连接数
    public static void main(String[] args) throws InterruptedException {
        Bootstrap bootstrap = new Bootstrap();
        bootstrap.group(new NioEventLoopGroup(2))
                .channel(NioSocketChannel.class)
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel ch) throws Exception {
                        ch.pipeline().addLast(TestHandler.INSTANT);
                    }
                });

        for (int i = 0; i < N; i++) {
            bootstrap.connect("localhost", 8080).sync();
        }
        System.out.println("已连接");

        new Thread(() -> {
            final long beginTime = System.currentTimeMillis();
            while(true) {
                long sum = TestHandler.SUM.get();
                int count = TestHandler.COUNT.get();
                if(count != 0) {
                    long duration = System.currentTimeMillis() - beginTime;
                    System.out.printf("QPS: %.2f, Response Time: %.2f\n",
                            1000F * count / duration,
                            sum / (float) count);
                }
                try {
                    Thread.sleep(2000);
                } catch (InterruptedException e) {
                    e.printStackTrace();
                }
            }
        }).start();
    }

    /**
     * 测试处理器
     */
    @ChannelHandler.Sharable
    private static class TestHandler extends ChannelInboundHandlerAdapter {

        /**
         * 总请求次数
         */
        public static final AtomicInteger COUNT = new AtomicInteger(0);
        /**
         * 总响应时间
         */
        public static final AtomicLong SUM = new AtomicLong(0);

        public static final TestHandler INSTANT = new TestHandler();

        @Override
        public void channelActive(ChannelHandlerContext ctx) throws Exception {
            ctx.executor().scheduleAtFixedRate(() -> {
                // 把当前时间发送到服务器去
                ByteBuf byteBuf = ctx.alloc().directBuffer(8);
                byteBuf.writeLong(System.currentTimeMillis());
                ctx.channel().writeAndFlush(byteBuf);
            }, 0, 1, TimeUnit.SECONDS);
        }

        private TestHandler() {}
        @Override
        public void channelRead(ChannelHandlerContext ctx, Object msg) throws Exception {
            ByteBuf byteBuf = (ByteBuf) msg;
            // 拿到时间
            long p = byteBuf.readLong();
            // 计算响应时间
            long time = System.currentTimeMillis() - p;
            // 累加次数
            COUNT.incrementAndGet();
            // 累加响应时间
            SUM.addAndGet(time);
        }
    }
}
